from pyflink.datastream import StreamExecutionEnvironment, ReduceFunction

env = StreamExecutionEnvironment.get_execution_environment()


env.set_parallelism(1)

students_ds = env.read_text_file('../../data/students.txt')

kv_ds = students_ds.map(lambda line: (line.split(',')[-1], 1))

key_ds = kv_ds.key_by(lambda kv: kv[0])

# key_ds.reduce(lambda a, b:(a[0],a[1]+b[1])).print()


class MyReduceClass(ReduceFunction):

    def reduce(self, value1, value2):
        return value1[0],value1[1]+value2[1]

key_ds.reduce(MyReduceClass()).print()
env.execute()